package com.hisense.ovcloud.spark.sql.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.stereotype.Component;
import reactor.kafka.receiver.ReceiverOptions;

import java.util.stream.Collectors;
import java.util.stream.Stream;

@Component
public class ReactiveConsumerConfig {

    @Value("${topics.deviceNickNameChange:deviceNickNameChange}")
    private String deviceNickNameChange;

    @Value("${topics.pairmessage:pairmessage}")
    private String pairmessage;

    @Value("${topics.modelConfigParam:model_config_param}")
    private String modelConfigParam;

    @Value("${topics.devicePowerFunctionMsg:device_power_function_msg}")
    private String devicePowerFunctionMsg;


    @Bean
    public ReceiverOptions<String, String> kafkaReceiverOptions(KafkaProperties kafkaProperties) {
        ReceiverOptions<String, String> basicReceiverOptions = ReceiverOptions.create(kafkaProperties.buildConsumerProperties());
        return basicReceiverOptions.subscription(Stream.of(deviceNickNameChange, pairmessage, modelConfigParam, devicePowerFunctionMsg).collect(Collectors.toList()));
    }

    @Bean
    public ReactiveKafkaConsumerTemplate<String, String> reactiveKafkaConsumerTemplate(ReceiverOptions<String, String> kafkaReceiverOptions) {
        return new ReactiveKafkaConsumerTemplate<>(kafkaReceiverOptions);
    }
}
